import com.zuikaku.pojo.AccessLogDO;
import com.zuikaku.pojo.ResultCount;
import com.zuikaku.source.AccessLogSource;
import com.zuikaku.util.TimeUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class AccessLogRealTimeMonitor {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置模式与并行数
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);
        //3.指定自定义source
        DataStreamSource<AccessLogDO> accessLogDS = env.addSource(new AccessLogSource());
        //4.过滤无用数据（url为空视为无用）
        DataStream<AccessLogDO> filterDS = accessLogDS.filter(new FilterFunction<AccessLogDO>() {
            @Override
            public boolean filter(AccessLogDO accessLogDO) throws Exception {
                return StringUtils.isNotEmpty(accessLogDO.getUrl());
            }
        });
        //5.定义waterMark指定事件时间，此数据流存在乱序延迟，接收3s内的延时数据
        DataStream<AccessLogDO> waterMarkDs = filterDS.assignTimestampsAndWatermarks(
                //接收3s内的延时数据
                WatermarkStrategy.<AccessLogDO>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        //指定事件时间列
                        .withTimestampAssigner((event, timestamp) -> {
                            return event.getCreateTime().getTime();
                        }));
        //6.定义最后兜底的侧向输出
        OutputTag<AccessLogDO> outputTag = new OutputTag<AccessLogDO>("lateData") {
        };

        //7.对url和httpCode进行分组
        KeyedStream<AccessLogDO, Tuple2<String, String>> keyedStream = waterMarkDs.keyBy(new KeySelector<AccessLogDO, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(AccessLogDO accessLogDO) throws Exception {
                return Tuple2.of(accessLogDO.getUrl(), String.valueOf(accessLogDO.getHttpCode()));
            }
        });

        //8.开窗，时间滑动窗口，使用事件时间EventTime，每5s(window slide)统计过去1min(window size)的数据
        WindowedStream<AccessLogDO, Tuple2<String, String>, TimeWindow> windowedStream = keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(60), Time.seconds(5)))
                //允许最大1min的延时数据
                .allowedLateness(Time.minutes(1))
                //指定最后兜底的侧向输出
                .sideOutputLateData(outputTag);
        //9.增量聚合：AggregateFunction与ProcessFunction混用，高级模式，可在增量聚合中拿到context上下文对象，从而获得更多信息
        SingleOutputStreamOperator<ResultCount> aggregate = windowedStream.aggregate(
                //输入对象是AccessLogDO，统计的是次数，所以ACC累加器类型为Long，输出也是统计的次数，也为Long
                new AggregateFunction<AccessLogDO, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;//累加器初始值为0
                    }

                    @Override
                    public Long add(AccessLogDO accessLogDO, Long old) {
                        return old + 1;//每次旧值+1
                    }

                    @Override
                    public Long getResult(Long acc) {
                        return acc;//返回累加器本身
                    }

                    @Override
                    public Long merge(Long a, Long acc1) {
                        return a + acc1;
                    }
                },
                //输入为上一个的out因此为long，输出ResultCount对象，key为string类型
                new ProcessWindowFunction<Long, ResultCount, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<String, String> key, ProcessWindowFunction<Long, ResultCount, Tuple2<String, String>, TimeWindow>.Context context, Iterable<Long> iterable, Collector<ResultCount> collector) throws Exception {
                        ResultCount resultCount = new ResultCount();
                        resultCount.setUrl(key.f0);
                        resultCount.setCode(Integer.parseInt(key.f1));
                        resultCount.setCount(iterable.iterator().next());//迭代器只有一个值，因为这是增量聚合，每次就一个
                        resultCount.setType("每5秒统计近1分接口PV");
                        resultCount.setStartTime(TimeUtil.format(context.window().getStart()));
                        resultCount.setEndTime(TimeUtil.format(context.window().getEnd()));
                        collector.collect(resultCount);
                    }
                });
        aggregate.print("统计后");

        //Pattern定义,连续两次不是200
        Pattern<AccessLogDO, AccessLogDO> pattern = Pattern.<AccessLogDO>begin("errorMonitor").where(new SimpleCondition<AccessLogDO>() {
            @Override
            public boolean filter(AccessLogDO accessLogDO) throws Exception {
                return accessLogDO.getHttpCode() != 200;
            }
        }).times(2);
        //匹配检查
        PatternStream<AccessLogDO> patternDs = CEP.pattern(keyedStream, pattern);

        //取出匹配流，指定输出结果
        SingleOutputStreamOperator<ResultCount> select = patternDs.select(new PatternSelectFunction<AccessLogDO, ResultCount>() {
            @Override
            public ResultCount select(Map<String, List<AccessLogDO>> map) throws Exception {
                List<AccessLogDO> accessLogDOS = map.get("errorMonitor");
                ResultCount resultCount = new ResultCount();
                resultCount.setCode(accessLogDOS.get(0).getHttpCode());
                resultCount.setCount((long) accessLogDOS.size());
                resultCount.setUrl(accessLogDOS.get(0).getUrl());

                return resultCount;
            }
        });
        //预警输出
        select.printToErr("预警");

        try {
            env.execute("AccessLogMonitor");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }
}
